package top.gcl.controller;

import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {
    // 消费监听
//    @KafkaListener(topics = {"userlog"})
//    public void onMessage(ConsumerRecord<?, ?> record) {
//        // 消费的哪个topic、partition的消息,打印出消息内容
//        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
//    }

//kafka教程-springboot消费者指定分区（partition）和偏移量（offset）
//    @KafkaListener(topicPartitions = {
//            @TopicPartition(topic = "userlog", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "0"))
//    })
//    public void onMessage(ConsumerRecord<?, ?> record) {
//        System.out.println("消费：topic:"+record.topic()+"-partition:"+record.partition()+"-offset:"+record.offset()+"-value:"+record.value());
//    }

//ack方式手动提交offset
//    @KafkaListener(topics = {"userlog"})
//    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
//        // 消费的哪个topic、partition的消息,打印出消息内容
//        System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", record.topic(),record.partition(),record.offset(),record.value()));
//        ack.acknowledge();
//    }

    //同步方式（没讲）commitSync方式
    //commitAsync 方式  异步方式
    @KafkaListener(id = "consumer1",topics = {"userlog"})
    public void onMessage(@Payload String message,
                          @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                          @Header(KafkaHeaders.OFFSET) String offset,
                          Consumer consumer) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", topic,partition,offset,message));
        consumer.commitAsync();
    }


}
